package producer

import (
	"gitee.com/jianbiao/go-util/netx"
	"gitee.com/jianbiao/go-util/tcppool"
	"github.com/pkg/errors"
	"github.com/streadway/amqp"
	"strconv"
	"sync"
	"time"
)

var rabbitPool tcppool.Pool
var once sync.Once
var pconfig *poolConfig

//配置
type poolConfig struct {
	Addr string
	QueueName string
	Durable bool
	InitialCap int
	MaxIdle int
	MaxCap int
	IdleTimeout time.Duration
	Confirm bool
}

//  定义一个消息队列结构体：helloworld 模型
type producer struct {
	queueName  string
	durable    bool
	Channel    *amqp.Channel
	close      func()
}

func initPool() error {
	//factory 创建连接的方法
	factory := func() (interface{}, error){
		return amqp.Dial(pconfig.Addr)
	}
	//close 关闭连接的方法
	close := func(v interface{}) error {
		return v.(*amqp.Connection).Close()
	}
	//ping
	ping := func(interface{}) error{
		uri, _ := amqp.ParseURI(pconfig.Addr)
		serverUrl := uri.Host + ":" + strconv.Itoa(uri.Port)
		_, err := netx.IsActiveTCPAddr(serverUrl)
		return err
	}
	//创建一个连接池： 初始化2，最大连接5，空闲连接数是4
	poolConfig := &tcppool.Config{
		InitialCap: pconfig.InitialCap,
		MaxIdle:    pconfig.MaxIdle,
		MaxCap:     pconfig.MaxCap,
		Factory:    factory,
		Close:      close,
		Ping:       ping,
		//连接最大空闲时间，超过该时间的连接 将会关闭，可避免空闲时连接EOF，自动失效的问题
		IdleTimeout: pconfig.IdleTimeout,
	}
	var err error
	rabbitPool, err = tcppool.NewChannelPool(poolConfig)
	if err != nil{
		return err
	}
	return nil
}

func newPoll(pc *poolConfig) (pro *producer, err error) {
	once.Do(func() {
		pconfig = pc
		err = initPool()
	})
	if err != nil{
		return nil, err
	}
	mq, err := rabbitPool.Get()
	if err != nil{
		return nil, err
	}
	mqconn := mq.(*amqp.Connection)
	ch, err := mqconn.Channel()
	if err != nil{
		return nil, err
	}
	//发送消息确认模式
	if pconfig.Confirm{
		err = ch.Confirm(false)
		if err != nil{
			return nil, err
		}
	}

	return &producer{
		queueName: pconfig.QueueName,
		durable:   pconfig.Durable,
		Channel:   ch,
		close: func() {
			ch.Close()
			rabbitPool.Put(mq)
		},
	}, nil
}

func (p *producer) publish(body string) (err error) {
	//确认模式
	if pconfig.Confirm{
		confirms := p.Channel.NotifyPublish(make(chan amqp.Confirmation, 1))
		defer func() {
			if err != nil{
				return
			}
			if confirmed := <-confirms; !confirmed.Ack {
				err = errors.New("发送消息失败")
			}
		}()
	}
	deliveryMode := amqp.Transient
	if pconfig.Durable {
		deliveryMode = amqp.Persistent
	}
	// 投递消息
	err = p.Channel.Publish(
		"",          // helloworld 、workqueue 模式设置为空字符串，表示使用默认交换机
		p.queueName, // 注意：简单模式 key 表示队列名称
		false,
		false,
		amqp.Publishing{
			DeliveryMode: deliveryMode,
			ContentType:  "text/plain",
			Body:         []byte(body),
		})
	return err
}

